package com.tcp2.net;

import com.google.common.collect.Maps;

import com.tcp2.server.DxsConfig;
import com.tcp2.server.MessageDispatcher;
import com.tcp2.server.ProtoConstants;
import com.tcp2.server.ProtoMsg;
import io.netty.channel.*;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Random;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Author: 杨俊辉
 * Time: 2014-09-23 10:02
 * Copyright (C) 2014 Xiamen Yaxon Networks CO.,LTD.
 */
@ChannelHandler.Sharable
public class MasterLinkHandler extends ChannelInboundHandlerAdapter {

    protected static Logger logger = LoggerFactory.getLogger(MasterLinkHandler.class);
    //获取下级平台ip-> 连接的的次数
    private static ConcurrentMap<String, LinkCount> ip2LinkCountMap = Maps.newConcurrentMap();

    //下级平台连接次数
    private class LinkCount {
        public long lastTime =0; //最后更新的时间
        public int count =0; //每次无效时间间隔内，累计最大连接次数（默认不大于15）
    }

    private  AttributeKey<Integer> platformAttrKey = AttributeKey.valueOf("platformKey"+new  Random().toString());
    private MasterLink masterLink;
    private MessageDispatcher messageDispatcher;
    private DxsConfig dxsConfig;
    private AtomicInteger snGen = new AtomicInteger(0);


    public MasterLinkHandler(MasterLink masterLink, MessageDispatcher messageDispatcher, DxsConfig dxsConfig) {
        this.masterLink = masterLink;
        this.messageDispatcher = messageDispatcher;
        this.dxsConfig = dxsConfig;

    }

    public MessageDispatcher getMessageDispatcher() {
        return messageDispatcher;
    }

    public void setMessageDispatcher(MessageDispatcher messageDispatcher) {
        this.messageDispatcher = messageDispatcher;
    }

    public DxsConfig getDxsConfig() {
        return dxsConfig;
    }

    public void setDxsConfig(DxsConfig dxsConfig) {
        this.dxsConfig = dxsConfig;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        try {
            //remoteAddress()获取的值"/127.0.0.1:50301"
            String remoteAddressStr = ctx.channel().remoteAddress().toString();
            String remoteAddressData[] = remoteAddressStr.split(":");
            String ip = remoteAddressData[0].substring(1, remoteAddressData[0].length());
            logger.info("[{}]主链路连接已建立",  ctx.channel().remoteAddress());
            if (ip != null && !ip.trim().equals("")) {
                long currentTime = System.currentTimeMillis();
                LinkCount linkCount = ip2LinkCountMap.get(ip);
                if(linkCount == null){
                    linkCount = new LinkCount();
                    ip2LinkCountMap.put(ip, linkCount);
                }
                if (linkCount.count == 0) {

                    linkCount.lastTime = currentTime;
                    linkCount.count++;
                } else {

                    logger.info("检测下级平台请求次数限制 start");

                    //距离上次连接的间隔时间
                    long intervalTime = currentTime - linkCount.lastTime;

                    //政府平台处理下级平台连接次数限制，默认在10秒内不能超过5次连接，屏蔽下级平台1个小时
                    if (intervalTime >= masterLink.getLimitConnectTimeInterval()) {
                        if (linkCount.count <= masterLink.getLimitConnectTimeIntervalMaxCount()) {
                            linkCount.count = 0;
                            linkCount.lastTime = currentTime;
                        } else {
                            if (intervalTime <= masterLink.getContinueConnectTimeInterval()) {
                                ctx.channel().close();
                            } else {
                                linkCount.count = 0;
                                linkCount.lastTime = currentTime;
                            }
                        }

                    } else {
                        if (linkCount.count <= masterLink.getLimitConnectTimeIntervalMaxCount()) {
                            linkCount.count++;
                        } else {
                            linkCount.lastTime = currentTime;
                            ctx.channel().close();
                        }
                    }

                    logger.info("检测下级平台请求次数限制 end");
                }
            }

        } catch (Exception e) {
            logger.info("{}主链路连接已建立失败", ctx.channel().remoteAddress(), e);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Integer gnssCenterId = ctx.channel().attr(platformAttrKey).get();
        logger.info("[{}]主链路连接已断开,下级平台接入码：{}", ctx.channel().remoteAddress(), gnssCenterId);
        if(gnssCenterId !=null){
            //关闭下级平台连接
            ConcurrentMap<Integer, PlatformChannel> plat2channelMap = PlatformMap.getPlat2channelMap();
            PlatformChannel platformChannel = plat2channelMap.get(gnssCenterId);
            if(platformChannel != null) {
                platformChannel.disconnected(ctx.channel());
                plat2channelMap.remove(gnssCenterId);
            }
            ConcurrentMap<Integer, SlaveLink> gov2platMap = PlatformMap.getGov2platMap();
            SlaveLink slaveLink = gov2platMap.get(gnssCenterId);
            if(slaveLink != null){
                slaveLink.stop();
                gov2platMap.remove(gnssCenterId);
            }
            masterLink.savePlatformLinkLog(gnssCenterId,(byte)0, (byte)2);
         }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.warn("主链路连接异常，即将关闭从链路连接", cause);
        ctx.close();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            ProtoMsg m = (ProtoMsg) msg;
            PlatformChannel plc = PlatformMap.getPlat2channelMap().get(m.gnssCenterId);
            if (m.msgId != ProtoConstants.UP_CONNECT_REQ && (plc != null &&
                    !plc.getChannel().equals(ctx.channel()))) {
                throw new RuntimeException("主链路未连接，消息id：" + m.msgId +
                        "，下级平台接入码：" + m.gnssCenterId);
            }

            ConcurrentMap<Integer, PlatformChannel> plat2channelMap = PlatformMap.getPlat2channelMap();
            if (plc == null) {
                plc = new PlatformChannel();
                Attribute<Integer> attr = ctx.channel().attr(platformAttrKey);
                attr.set(m.gnssCenterId);
                plc.getLastContactTime().set(System.currentTimeMillis());
                plc.setChannel(ctx.channel());
            } else {
                plc.setChannel(ctx.channel());
            }
            plat2channelMap.put(m.gnssCenterId, plc);
            PlatformMap.setPlat2channelMap(plat2channelMap);

            ConcurrentMap<Integer, MasterLinkHandler> plat2govMap = PlatformMap.getPlat2govMap();
            plat2govMap.put(m.gnssCenterId, this);
            PlatformMap.setPlat2govMap(plat2govMap);
           messageDispatcher.receiveMessage(m);

        } catch (Exception e) {
            logger.info("处理主链路的上行消息异常", e);
        }
    }

    /**
     * 获取下一个流水号
     *
     * @return
     */
    private int nextSn() {
        return snGen.incrementAndGet();
    }

    /**
     * 向下级平台发送应答消息
     * @param channel 下级平台TCP连接通道
     * @param msg 发送的消息
     * @return
     */
    protected ChannelFuture sendMessage(Channel channel, ProtoMsg msg) {
        msg.msgSn = nextSn();
        return channel.writeAndFlush(msg);
    }




}
